package plgmqtt

import (
	"crypto/tls"
	"dgo/goutils/utils"
	"fmt"
	"gitee.com/ymofen/supervalue"
	mqtt "github.com/eclipse/paho.mqtt.golang"
	"net/url"
	"strings"
	"sync"
	"time"
)

/*
**

连接->订阅->拔网线: 不会重连
连接->订阅->拔网线->发布: 会重连

	  该类实现的功能：
	    1.首次连接失败后，会继续进行连接
	    2.连接丢失后，会重新进行订阅

	  mqtt库：
	    1. 首次连接失败后不会自动重连(断开网络，进行连接， 插上网线后，并进行连接)
	    2. 拔掉网线后，会执行OnConnectionLost
	    3. 连接丢失后，重新建立连接，需要重新订阅。
	    4. 相同的主题订阅两次，只会回调最后一次订阅的事件
	    5. 如果订阅a/#, a/b/+ 两个主题, 如果推送 a/b/c 的主题， 则每个订阅会响应两次, 库的(BUG)
	    5.1. 如果订阅a/#, a/b/+, a/b/c 3个主题, 如果推送 a/b/c 的主题， 则每个订阅会响应3次, 库的(BUG)
				inner.SubscribeEx("a/b/+", func(topic string, buf []byte) {
					t.Logf("1:topic:%s, data:%s", topic, string(buf))
				})

				inner.SubscribeEx("a/#", func(topic string, buf []byte) {
					t.Logf("2:topic:%s, data:%s", topic, string(buf))
				})

				inner.SubscribeEx("a/b/c", func(topic string, buf []byte) {
					t.Logf("3:topic:%s, data:%s", topic, string(buf))
				})

				time.AfterFunc(time.Second, func() {
					inner.Publish("a/b/c", []byte("hello"))
				})
*/
type MqttInner struct {
	connectDNA     int32
	lk             sync.RWMutex
	opts           *mqtt.ClientOptions
	clt            mqtt.Client
	clientid       string
	addr           string
	user           string
	pass           string
	status         string
	connect_status int8 // 0:未连接， 1：建立连接成功, 2:丢失连接

	sublst        []string
	OnMessage     func(sender *MqttInner, channelid string, data []byte)
	OnConnected   func(sender *MqttInner)
	OnConnectFail func(sender *MqttInner, err error)
	OnDisconnect  func(sender *MqttInner, err error)
}

func NewMqttInner() *MqttInner {
	rval := &MqttInner{}
	return rval
}

func (this *MqttInner) StatusString() string {
	return this.status
}

func (this *MqttInner) SubTopics(sep string) string {
	return strings.Join(this.sublst, sep)
}

func (this *MqttInner) onMqttMsg(client mqtt.Client, msg mqtt.Message) {
	if utils.GoFunCatchException {
		defer utils.PanicHandler()
	}
	buf := msg.Payload()
	evt := this.OnMessage
	if evt != nil {
		evt(this, msg.Topic(), buf)
	}
}

/*
*

	马上进行推送
*/
func (this *MqttInner) Publish0(topic string, qos byte, retained bool, buf []byte) error {
	if utils.GoFunCatchException {
		defer utils.PanicHandler()
	}
	clt := this.clt
	if clt == nil {
		return fmt.Errorf("尚未建立连接")
	}
	token := clt.Publish(topic, qos, retained, buf)
	if token.WaitTimeout(time.Second * 10) {
		return token.Error()
	}
	return fmt.Errorf("取消响应超时")
}

/*
*

	马上进行推送
*/
func (this *MqttInner) Publish(topic string, buf []byte) error {
	return this.Publish0(topic, 0, false, buf)
}

/*
*

	同一个主题只能订阅一次， 只回调最后一个函数
*/
func (this *MqttInner) SubscribeEx(topic string, onMessage func(topic string, buf []byte)) error {
	clt := this.clt
	if clt == nil {
		return fmt.Errorf("尚未建立连接")
	}
	token := clt.Subscribe(topic, 0, func(client mqtt.Client, msg mqtt.Message) {
		buf := msg.Payload()
		onMessage(msg.Topic(), buf)
	})
	if token.WaitTimeout(time.Second * 10) {
		return token.Error()
	}
	return fmt.Errorf("订阅响应超时")
}

/*
**

	马上订阅
*/
func (this *MqttInner) innerSubscribe(topic string) error {
	clt := this.clt
	if clt == nil {
		return fmt.Errorf("尚未建立连接")
	}
	token := clt.Subscribe(topic, 0, this.onMqttMsg)
	if token.WaitTimeout(time.Second * 10) {
		return token.Error()
	}
	return fmt.Errorf("订阅响应超时")
}

/*
**

	取消马上订阅
*/
func (this *MqttInner) innerUnsubscribe(topic string) {
	clt := this.clt
	if clt == nil {
		return
	}
	clt.Unsubscribe(topic)
}

/*
*

	以topics为主
*/
func (this *MqttInner) Resubscribe(topics string) {
	this.lk.Lock()
	defer this.lk.Unlock()
	strs := strings.Split(topics, ",")

	connect_status := this.connect_status
	// 找出需要去掉订阅的
	for i := len(this.sublst) - 1; i >= 0; i-- {
		if utils.StrIndex(this.sublst[i], strs...) == -1 { // 新的不存在
			if connect_status == 1 {
				this.innerUnsubscribe(this.sublst[i])
			}
			this.sublst = this.sublst[:i] // 删除掉最后一个
		}
	}

	var err error
	for i := 0; i < len(strs); i++ {
		if utils.StrIndex(strs[i], this.sublst...) == -1 { // 新的不存在
			this.sublst = append(this.sublst, strs[i])
			if connect_status == 1 && err == nil {
				err = this.innerSubscribe(this.sublst[i])
			}
		}
	}
	if err != nil { // 订阅出错，重新订阅
		this.RequestDis(err.Error())
	}
}

/*
*
 */
func (this *MqttInner) Subscribe(topic string) {
	this.lk.Lock()
	defer this.lk.Unlock()
	if utils.StrIndex(topic, this.sublst...) == -1 {
		this.sublst = append(this.sublst, topic)
		if this.connect_status == 1 { // 连接状态
			this.innerSubscribe(topic)
		}
	}
}

func (this *MqttInner) Unsubscribe(topic string) {
	this.lk.Lock()
	defer this.lk.Unlock()
	idx := utils.StrIndex(topic, this.sublst...)
	if idx == -1 { // 没有订阅
		return
	}
	this.innerUnsubscribe(topic)
	this.sublst = append(this.sublst[:idx], this.sublst[idx+1:]...)

}

func (this *MqttInner) RequestDis(reason string) {
	clt := this.clt
	if clt != nil {
		clt.Disconnect(1)
		clt = nil
	}
}

func (this *MqttInner) Close() {
	this.connectDNA++
	this.RequestDis("执行关闭")
}

func (this *MqttInner) innerConnectFail(err error) {
	evt := this.OnConnectFail
	if evt != nil {
		evt(this, err)
	}
}

func (this *MqttInner) innerConnect() {
	dna := this.connectDNA
	// 首次进行确保连接成功
	utils.WaitTimeOut(time.Second*10, func() {
		clt := mqtt.NewClient(this.opts)
		for { // 持续连接
			if dna != this.connectDNA {
				this.status = fmt.Sprintf("[%s]本次连接[%d]中断", utils.NowStringEx(), dna)
				break
			}
			token := clt.Connect()
			token.WaitTimeout(time.Second * 5)
			err := token.Error()
			if err != nil {
				this.status = fmt.Sprintf("[%s]连接失败,%s", utils.NowStringEx(), err.Error())
				this.innerConnectFail(err)
				time.Sleep(time.Second * 10)
			} else {
				this.clt = clt
				break // 连接完成
			}
		}
	})
}

/*
*

	{
	    "clientid":"",
		"addr":"tcp://139.199.7.76:52883",
	    "user":"",
	    "pass":"",
	}
*/
func (this *MqttInner) ReloadConf(conf *wrapper.SuperValue) {

	isChanged := false
	conf.CheckExistsStrFunc("addr", func(strv string) {
		if strv != this.addr {
			this.addr = strv
			isChanged = true
		}
	})

	conf.CheckExistsStrFunc("host", func(strv string) {
		if strv != this.addr {
			this.addr = strv
			isChanged = true
		}
	})

	clientid := conf.StringByName("clientid", "")
	if this.clientid != clientid {
		this.clientid = clientid
		isChanged = true
	}

	user := conf.StringByName("user", "")
	if this.user != user {
		this.user = user
		isChanged = true
	}

	pass := conf.StringByName("pass", "")
	if this.pass != pass {
		this.pass = pass
		isChanged = true
	}

	if isChanged {
		// 重新新的任务
		this.connectDNA++
		this.RequestDis("配置改变")
		if len(this.addr) > 0 {
			opts := mqtt.NewClientOptions()
			opts.AddBroker(this.addr).SetAutoReconnect(true)
			opts.SetKeepAlive(time.Second * 10)
			if len(this.user) > 0 {
				opts.SetUsername(this.user).SetPassword(this.pass)
			}
			if len(this.clientid) > 0 {
				opts.SetClientID(this.clientid)
			}
			opts.SetConnectionLostHandler(func(client mqtt.Client, err error) {
				this.connect_status = 2
				evt := this.OnDisconnect
				if evt != nil {
					evt(this, err)
				}
				this.status = fmt.Sprintf("[%s]丢失连接:%s", utils.NowStringEx(), err.Error())
			})

			opts.SetOnConnectHandler(func(client mqtt.Client) {
				if this.OnConnected != nil {
					this.OnConnected(this)
				}
				this.status = fmt.Sprintf("[%s]连接成功(%s)", utils.NowStringEx(), this.addr)
				this.connect_status = 1

				// 重新进行订阅
				this.lk.RLock()
				defer this.lk.RUnlock()
				for i := 0; i < len(this.sublst); i++ {
					topic := utils.Trim(this.sublst[i])
					if len(topic) > 0 {
						err := this.innerSubscribe(this.sublst[i])
						if err != nil {
							go func() {
								if utils.GoFunCatchException {
									defer utils.PanicHandler()
								}
								this.RequestDis(fmt.Sprintf("[%s]订阅失败, 断开连接", topic))
							}()
							break
						}
					}
				}
			})
			opts.SetConnectionAttemptHandler(func(broker *url.URL, tlsCfg *tls.Config) *tls.Config {
				this.status = fmt.Sprintf("[%s]尝试进行连接(%s)", utils.NowStringEx(), broker.String())
				return tlsCfg
			})
			this.opts = opts
			this.innerConnect()
		}
	}
}
